package com.future.myjavatools.core.zhoupu;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 去重队列
 *
 * @param <T>
 */
public class AntiDuplicateBlockingQueue<T> {

    private final Queue<T> queue;
    private final Set<T> deDupSet;

    private final ReentrantLock putLock = new ReentrantLock();
    private final ReentrantLock takeLock = new ReentrantLock();

    public AntiDuplicateBlockingQueue(int capacity) {
        queue = new LinkedBlockingQueue<T>(capacity);
        deDupSet = Sets.newConcurrentHashSet();
    }

    public boolean offer(T item) {
        if (item == null) {
            throw new NullPointerException();
        }
        if (deDupSet.contains(item)) {
            return true;
        }
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (!deDupSet.contains(item) && queue.offer(item)) {
                return deDupSet.add(item);
            }
        } finally {
            putLock.unlock();
        }
        return false;
    }

    public T poll() {
        if (queue.size() == 0) {
            return null;
        }
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            T item = queue.peek();
            if (item != null && deDupSet.remove(item)) {
                return queue.poll();
            }
        } finally {
            takeLock.unlock();
        }
        return null;
    }

    public int size() {
        return queue.size();
    }

    public List<T> dump() {
        return Collections.unmodifiableList(Lists.newArrayList(deDupSet));
    }

}
